package pro4_threadpattern.nettyReactor147;

import io.netty.bootstrap.ServerBootstrap;
import io.netty.buffer.ByteBuf;
import io.netty.buffer.Unpooled;
import io.netty.channel.*;
import io.netty.channel.nio.NioEventLoopGroup;
import io.netty.channel.socket.nio.NioServerSocketChannel;
import io.netty.util.CharsetUtil;

/**
 * 1. 单reactor  多个worker线程模式不会干扰accept.
 *
 * @program: netty
 * @description: 单reactor    工作线程多个
 * @author: zy
 * @create: 2025-04-13 15:17
 */
public class Netty2_SingleReactorMultiWorkThreadServer {
    //获取cpu核数，作为线程池的大小
    //private static final int NTHREADS = Runtime.getRuntime().availableProcessors() * 2;
    private static final int NTHREADS =5;

    public static void main(String[] args) {
        EventLoopGroup reactorGroup = new NioEventLoopGroup(1);   //单reactor线程  boss线程组  ->解决版本一中boss线程的阻塞问题

        EventLoopGroup workerGroup = new NioEventLoopGroup(NTHREADS);   //多worker线程 执行业务逻辑

        try {
            ServerBootstrap serverBootstrap = new ServerBootstrap();
            serverBootstrap.group(reactorGroup, workerGroup)    // 和accept  and IO   使用同一个线程组
                    .channel(NioServerSocketChannel.class)
                    .childHandler(new ChannelInitializer<>() {
                        @Override
                        protected void initChannel(Channel ch) throws Exception {
                            //                 专门用于业务逻辑操作的 workerGroup
                            ch.pipeline().addLast(new SimpleServerHandler());
                            //将业务处理类(耗时的操作)放到了worker线程组中执行.
                        }
                    });
            ChannelFuture future = serverBootstrap.bind(8888).sync();  // 绑定端口
            future.channel().closeFuture().sync();  // 关闭通道
        }catch(Exception ex){
            ex.printStackTrace();
        }finally{
            reactorGroup.shutdownGracefully();
            workerGroup.shutdownGracefully();
        }
    }

    //处理类
    private static class SimpleServerHandler extends ChannelInboundHandlerAdapter {
        // 连接事件
        @Override
        public void channelActive(ChannelHandlerContext ctx) throws Exception {
            System.out.println( ctx.channel().remoteAddress()+"  "+ctx.channel().eventLoop().toString()+", 连接事件"); //连接不会阻塞了, 但任务会放到worker线程中执行，还是有阻塞.
        }
        //事件处理的回调方法
        //读事件
        @Override
        public void channelRead(ChannelHandlerContext ctx, Object msg) throws Exception {

            Thread.sleep(  5000 );
            ByteBuf in=(ByteBuf)msg;  //读到的数据
            System.out.println("服务端收到数据: "+in.toString(   CharsetUtil.UTF_8     ));
            ctx.writeAndFlush(  Unpooled.copiedBuffer("hello client\n", CharsetUtil.UTF_8)  );  // \n是用于给客户端确认处理完毕.
        }
        @Override
        public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) throws Exception {
            cause.printStackTrace();
            ctx.close();
        }
    }
}
